package com.summer.rx.impl;

import java.util.concurrent.DelayQueue;
import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;

/**
 * 程序工人
 * @param <T> 处理数据类型
 */
public class Worker<T>{

	private static final Handler handler = new Handler();
	
    private final long period;
    private final TimeUnit unit;
    
    
    private Consumer<T> action;
    

    public Worker(long period, TimeUnit unit) {
        this.period = period;
        this.unit = unit;
    }
    

    public Worker<T> onRun(Consumer<T> action) {
    	this.action = action;
    	return this;
    }

    void subscribe(Consumer<T> consumer) {
    	consumer.accept(null);
    }

    public Worker<T> just(final T... data) {
    	return new Worker<T>(period, unit) {
    		@Override
            void subscribe(final Consumer<T> consumer) {
                if(data != null) {
                    for(T value : data) {
                    	consumer.accept(value);
                    }
                }
            }
    	};
    }
    
    public Worker<T> subscribeOn() {
        return new Worker<T>(period, unit) {
            @Override
            void subscribe(final Consumer<T> consumer) {
                Consumer<T> newConsumer = new Consumer<T>() {
                    @Override
                    public void accept(T value) {
                    	if(action != null) {
                    		action.accept(value);
                    	}
                        new Thread(() -> consumer.accept(value)).start();
                    }
                };
                Worker.this.<T>subscribe(newConsumer);
            }
        };
    }

    public Worker<T> subscribeBy() {
        return new Worker<T>(period, unit) {
            @Override
            void subscribe(final Consumer<T> consumer) {
                Consumer<T> newConsumer = new Consumer<T>() {
                    @Override
                    public void accept(T value) {
                    	if(action != null) {
                    		action.accept(value);
                    	}
                        handler.post(() -> consumer.accept(value));
                    }
                };
                Worker.this.<T>subscribe(newConsumer);
            }
        };
    }

    public void accept() {
    	if(action == null) {
    		throw new RuntimeException("Worker 至少需要一个 action 才可以运行，请先 onRun 指定一个行为");
    	}
    	
    	subscribe(new Consumer<T>() {
            @Override
            public void accept(T value) {
            	action.accept(value);
            	if(period >= 0) {
            		handler.post(() -> Worker.this.accept(), (int) unit.toMillis(period));
            	}
            }
        });
    }
    
    private static class Handler implements Runnable {
    	private DelayQueue<DelaySendTask> queue = new DelayQueue<>();
    	
    	public Handler() {
    		new Thread(this, "Handler").start();
    	}
    	
    	public void post(Runnable run) {
    		post(run, 0);
    	}
    	
    	public void post(Runnable run, int delayTime) {
    		queue.add(new DelaySendTask(run, delayTime));
    	}

		@Override
		public void run() {
			while (true) {
	            try {
	                DelaySendTask take = queue.take();
	                if (take != null) {
	                    take.execTask();
	                }
	            } catch (InterruptedException e) {
	                e.printStackTrace();
	            }
	        }
		}
    	
		private class DelaySendTask implements Delayed {
	        private final long availableTime;
	        private final Runnable runnable;

	        public DelaySendTask(Runnable runnable, int delayTime) {
	            this.runnable = runnable;
	            this.availableTime = delayTime + System.currentTimeMillis();
	        }

	        public void execTask() {
	            if(runnable != null) {
	            	runnable.run();
	            }
	        }

	        @Override
	        public long getDelay(TimeUnit timeUnit) {
	            //判断 availableTime 是否大于当前系统时间，并将结果转换成 MILLISECONDS
	            long diffTime = availableTime - System.currentTimeMillis();
	            return timeUnit.convert(diffTime, TimeUnit.MILLISECONDS);
	        }

	        @Override
	        public int compareTo(Delayed delayed) {
	            // compareTo 用在 DelaySendTask 的排序
	            return (int) (this.availableTime - ((DelaySendTask) delayed).availableTime);
	        }
	    }
    }
    
}